package com.mangix.scheduler;

import com.mangix.core.MangixNode;
import com.mangix.files.MangixFiles;
import com.mangix.http.MangixHttpClient;
import com.mangix.http.MangixHttpHeader;
import com.mangix.utils.MangixUtils;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.client.util.PathContentProvider;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Replicate the object to other nodes.
 * Created by ZXL on 2014/6/11.
 */
public class MangixReplicator extends Thread {

    private ConcurrentHashMap<Path, Boolean> taskMap;
    private BlockingQueue<Path> taskQueue;
    private MangixNode localNode;
    private int M;
    private MangixFiles mangixFiles;

    public MangixReplicator(ConcurrentHashMap<Path, Boolean> taskMap, BlockingQueue<Path> taskQueue, MangixNode localNode) {
        this.taskMap = taskMap;
        this.taskQueue = taskQueue;
        this.localNode = localNode;
        this.M = localNode.getConfig().getM();
        this.mangixFiles = localNode.getFiles();
    }

    private void replicate(MangixNode dest, Path file, String key) throws IOException {
        try {
            MangixHttpClient.put("http://" + dest.getHostName() + ":" + dest.getPort() + "/direct/objects/" + key, file, file.getFileName().toString(), localNode.getFiles().getReplicasNum(file));
            System.out.println(localNode.toString() + " replicate: " + file.toString() + " successful!");
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println(localNode.toString() + " replicate: " + file.toString() + " failed!");
        }

    }

    private boolean delete(MangixNode dest, Path file, String key) throws IOException {
        try {
            MangixHttpClient.delete("http://" + dest.getHostName() + ":" + dest.getPort() + "/direct/objects/" + key + "?delete_file=true", file.getFileName().toString());
            return true;
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println(dest.toString() + " delete: " + file.toString() + " failed!");
            return false;
        }
    }

    public void run() {
        Path file;
        try {
            while ((file = taskQueue.take()) != null) {
                try {
                    String key = file.getParent().getFileName().toString();
                    Long id = MangixUtils.hash(key, M);
                    ArrayList<MangixNode> nodeList = new ArrayList<>();
                    ArrayList<MangixNode> predecessorList = localNode.getPredecessorList();
                    ArrayList<MangixNode> successorList = localNode.getSuccessorList();
                    int replicas = mangixFiles.getReplicasNum(file);
                    // Put the predecessor list, localNode node, successor list to the same list, in the order in chord ring.
                    for (int i = (predecessorList.size() < replicas)? (predecessorList.size() - 1): (replicas - 1); i >= 0; i--)
                        nodeList.add(predecessorList.get(i));
                    nodeList.add(localNode);
                    for (int i = 0; i < replicas && i < successorList.size(); i++)
                        nodeList.add(successorList.get(i));
                    // Find the first coordinator of this key.
                    int offset = -1;
                    for (int i = 0; i < nodeList.size() - replicas; i++) {
                          if (MangixUtils.isIDInInterval(id, nodeList.get(i).hash(), nodeList.get(i + 1).hash(), false, true, M)) {
                            offset = i;
                            break;
                        }
                    }
                    if (offset != -1) {
                        // Try to delete old object in remote nodes.
                        if(file.getFileName().toString().split("\\.")[1].equals("delete")) {
                            boolean isDeleteSuccess = true;
                            for (int i = offset + 1; i < nodeList.size() && i <= offset + replicas; i++) {
                                MangixNode dest = nodeList.get(i);
                                if (dest == localNode) continue;
                                if (!delete(dest, file, key)) {
                                    isDeleteSuccess = false;
                                }
                            }
                            if (isDeleteSuccess) {
                                Files.deleteIfExists(file);
                                System.out.println("Node " + localNode.toString() + " delete file " + file.toString() + " success.");
                            }
                        } else {
                            // Try to replicate this object to other nodes.
                            for (int i = offset + 1; i < nodeList.size() && i <= offset + replicas; i++) {
                                MangixNode dest = nodeList.get(i);
                                if (dest == localNode) continue;
                                Map meta = dest.getMeta(key);
                                if (meta == null) {
                                    replicate(dest, file, key);
                                } else {
                                    long localVersion = Long.parseLong(file.getFileName().toString().split("\\.")[0]);
                                    long remoteVersion = (Long)meta.get("version");
                                    if (localVersion < remoteVersion) {
                                        Files.deleteIfExists(file);
                                    } else if (localVersion > remoteVersion) {
                                        System.out.println("Try to replicate file: " + file.toString() + ", local version is " + localVersion + ", remote version is " + remoteVersion);
                                        replicate(dest, file, key);
                                    }
                                }
                            }
                        }
                    }
                } catch (NoSuchFileException e) {
                    // Do nothing.
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    taskMap.remove(file);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
